/*
 * Copyright (c) 2025 CAX Conversion Project
 */
package com.comac.ins.isfd.service.impl;


import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.core.toolkit.Wrappers;
import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
import com.comac.ins.common.core.exception.base.BaseException;
import com.comac.ins.common.core.utils.MapstructUtils;
import com.comac.ins.common.core.utils.StringUtils;
import com.comac.ins.common.mybatis.core.page.PageQuery;
import com.comac.ins.common.mybatis.core.page.TableDataInfo;
import com.comac.ins.isfd.constant.enums.IsfdDatasourceSyncTypeEnum;
import com.comac.ins.isfd.constant.enums.IsfdDatasourceTaskSchedulingConfigStatusEnum;
import com.comac.ins.isfd.constant.enums.IsfdDatasourceTaskSchedulingConfigTypeEnum;
import com.comac.ins.isfd.constant.enums.IsfdDatasourceTaskStatusEnum;
import com.comac.ins.isfd.domain.IsfdDatasourceTaskInstanceLog;
import com.comac.ins.isfd.domain.IsfdDatasourceTaskSchedulingConfig;
import com.comac.ins.isfd.domain.bo.IsfdDatasourceTaskSchedulingConfigActionBo;
import com.comac.ins.isfd.domain.bo.IsfdDatasourceTaskSchedulingConfigBo;
import com.comac.ins.isfd.domain.vo.IsfdDatasourceDataIntegrationTaskVo;
import com.comac.ins.isfd.domain.vo.IsfdDatasourceTaskSchedulingConfigVo;
import com.comac.ins.isfd.mapper.IsfdDatasourceTaskInstanceLogMapper;
import com.comac.ins.isfd.mapper.IsfdDatasourceTaskSchedulingConfigMapper;
import com.comac.ins.isfd.service.IIsfdDatasourceDataIntegrationTaskService;
import com.comac.ins.isfd.service.IIsfdDatasourceTaskSchedulingConfigService;
import com.comac.ins.isfd.util.DateUtils;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Lazy;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.util.CollectionUtils;

import java.util.Date;
import java.util.List;
import java.util.Map;

/**
 * 数据源数据集成任务调度配置Service业务层处理
 *
 * @author hxloongs
 * @date 2024-10-11
 */
@Slf4j
@RequiredArgsConstructor
@Service
public class IsfdDatasourceTaskSchedulingConfigServiceImpl implements IIsfdDatasourceTaskSchedulingConfigService {

    @Autowired
    private IsfdDatasourceTaskSchedulingConfigMapper baseMapper;
    @Autowired
    private IsfdDatasourceTaskInstanceLogMapper isfdDatasourceTaskInstanceLogMapper;
    @Autowired
    private ThreadPoolTaskExecutor threadPoolTaskExecutor;
    @Lazy
    @Autowired
    private IIsfdDatasourceDataIntegrationTaskService iIsfdDatasourceDataIntegrationTaskService;

    /**
     * 查询数据源数据集成任务调度配置
     *
     * @param id 主键
     * @return 数据源数据集成任务调度配置
     */
    @Override
    public IsfdDatasourceTaskSchedulingConfigVo queryById(Long id) {
        return baseMapper.selectVoById(id);
    }

    /**
     * 分页查询数据源数据集成任务调度配置列表
     *
     * @param bo        查询条件
     * @param pageQuery 分页参数
     * @return 数据源数据集成任务调度配置分页列表
     */
    @Override
    public TableDataInfo<IsfdDatasourceTaskSchedulingConfigVo> queryPageList(IsfdDatasourceTaskSchedulingConfigBo bo, PageQuery pageQuery) {
        LambdaQueryWrapper<IsfdDatasourceTaskSchedulingConfig> lqw = buildQueryWrapper(bo);
        Page<IsfdDatasourceTaskSchedulingConfigVo> result = baseMapper.selectVoPage(pageQuery.build(), lqw);
        return TableDataInfo.build(result);
    }

    /**
     * 查询符合条件的数据源数据集成任务调度配置列表
     *
     * @param bo 查询条件
     * @return 数据源数据集成任务调度配置列表
     */
    @Override
    public List<IsfdDatasourceTaskSchedulingConfigVo> queryList(IsfdDatasourceTaskSchedulingConfigBo bo) {
        LambdaQueryWrapper<IsfdDatasourceTaskSchedulingConfig> lqw = buildQueryWrapper(bo);
        return baseMapper.selectVoList(lqw);
    }

    @Override
    public IsfdDatasourceTaskSchedulingConfigVo save(IsfdDatasourceTaskSchedulingConfigBo bo) {
        IsfdDatasourceTaskSchedulingConfig po = MapstructUtils.convert(bo, IsfdDatasourceTaskSchedulingConfig.class);
        if (IsfdDatasourceTaskSchedulingConfigTypeEnum.CYCLE.getValue().equals(bo.getSchedulingType())) {
            po.setTriggerTime(bo.getStartTime());
        } else {
            po.setTriggerTime(null);
        }
        po.setStatus(IsfdDatasourceTaskSchedulingConfigStatusEnum.NOT_RUNNING.getValue());
        //新增或者修改
        if (bo.getId() == null) {
            baseMapper.insert(po);
        } else {
            baseMapper.updateById(po);
        }
        return MapstructUtils.convert(po, IsfdDatasourceTaskSchedulingConfigVo.class);
    }

    @Override
    public List<IsfdDatasourceTaskSchedulingConfigVo> scanBeforeNowTimeList(Date nowTime) {
        return baseMapper.selectVoList(new LambdaQueryWrapper<IsfdDatasourceTaskSchedulingConfig>()
            .ne(IsfdDatasourceTaskSchedulingConfig::getTriggerTime, null)
            .le(IsfdDatasourceTaskSchedulingConfig::getTriggerTime, nowTime)
            .ge(IsfdDatasourceTaskSchedulingConfig::getEndTime, nowTime)
            .eq(IsfdDatasourceTaskSchedulingConfig::getStatus, IsfdDatasourceTaskSchedulingConfigStatusEnum.NOT_RUNNING.getValue()));
    }

    @Override
    public Date findTriggerTime(Date nowTime, Long taskSchedulingConfigId) {
        IsfdDatasourceTaskSchedulingConfig po = baseMapper.selectById(taskSchedulingConfigId);
        Long schedulingCycle = po.getSchedulingCycle();
        String cycleUnit = po.getCycleUnit();
        Date startTime = po.getStartTime();
        Date endTime = po.getEndTime();
        Date triggerTime = null;
        // 将时间段切分成多个时间点
        List<Date> timePoints = DateUtils.getTimePoints(startTime, endTime, schedulingCycle, cycleUnit);
        if (!CollectionUtils.isEmpty(timePoints)) {
            // 查找最近的且大于 nowTime 的时间点
            triggerTime = DateUtils.findNearestLaterTimePoint(timePoints, nowTime);
        }
        if (triggerTime == null) {
            log.info("reset trigger time is null ,now time is {} ,taskSchedulingConfigId is {}", nowTime, taskSchedulingConfigId);
        }
        return triggerTime;
    }

    @Transactional
    @Override
    public Boolean suspend(IsfdDatasourceTaskSchedulingConfigActionBo actionBo) {
        Long taskId = actionBo.getTaskId();
        IsfdDatasourceDataIntegrationTaskVo taskVo = iIsfdDatasourceDataIntegrationTaskService.queryById(taskId);
        if (taskVo == null || IsfdDatasourceTaskStatusEnum.RUNNING.getValue() != taskVo.getStatus()) {
            throw new BaseException("任务不存在或者状态不正确，请确认后重试！");
        }

        if (!IsfdDatasourceSyncTypeEnum.REAL_TIME_SYNC.getValue().equals(taskVo.getSyncType())) {
            throw new BaseException("非实时任务不能中止，请确认后重试！");
        }

        // todo 调用seatunnel 接口进行任务中止

        // 修改任务状态
        iIsfdDatasourceDataIntegrationTaskService.updateStatusAndTriggerTime(IsfdDatasourceTaskStatusEnum.TERMINATED.getValue(), taskId, null);
        return Boolean.TRUE;
    }

    @Transactional
    @Override
    public Boolean trigger(IsfdDatasourceTaskSchedulingConfigActionBo actionBo) {
        // 当前时间作为触发时间
        Date nowTime = new Date();
        Long taskId = actionBo.getTaskId();
        IsfdDatasourceDataIntegrationTaskVo taskVo = iIsfdDatasourceDataIntegrationTaskService.queryById(taskId);
        if (taskVo == null) {
            throw new BaseException("任务不存在或者状态不正确，请确认后重试！");
        }

        IsfdDatasourceTaskStatusEnum isfdDatasourceTaskStatusEnum = null;
        // 如果是离线任务
        if (IsfdDatasourceSyncTypeEnum.OFFLINE_SYNC.getValue().equals(taskVo.getSyncType())) {
            IsfdDatasourceTaskSchedulingConfig taskSchedulingConfig = baseMapper.selectOne(new LambdaQueryWrapper<IsfdDatasourceTaskSchedulingConfig>()
                .eq(IsfdDatasourceTaskSchedulingConfig::getTaskId, taskId)
                .eq(IsfdDatasourceTaskSchedulingConfig::getStatus, IsfdDatasourceTaskSchedulingConfigStatusEnum.NOT_RUNNING.getValue()));
            if (taskSchedulingConfig == null) {
                throw new BaseException("调度配置不存在或者状态不正确，请确认后重试！");
            }
            if (nowTime.after(taskSchedulingConfig.getEndTime()) || nowTime.before(taskSchedulingConfig.getStartTime())) {
                throw new BaseException("当前时间大于调度结束时间，或者小于调度开始时间，请确认后重试！");
            }
            //修改离线任务的调度配置状态
            taskSchedulingConfig.setStatus(IsfdDatasourceTaskSchedulingConfigStatusEnum.RUNNING.getValue());
            baseMapper.updateById(taskSchedulingConfig);
            isfdDatasourceTaskStatusEnum = IsfdDatasourceTaskStatusEnum.SCHEDULING;
        } else {
            isfdDatasourceTaskStatusEnum = IsfdDatasourceTaskStatusEnum.RUNNING;
        }

        //修改任务状态
        iIsfdDatasourceDataIntegrationTaskService.updateStatusAndTriggerTime(isfdDatasourceTaskStatusEnum.getValue(), taskId, nowTime);
        //调用工具，手动外部触发需要异步，定时任务扫描的不需要异步
        if (actionBo.getNeedSync()) {
            threadPoolTaskExecutor.execute(() -> {
                this.useSeaTunnelTool(taskVo, nowTime);
            });
        } else {
            this.useSeaTunnelTool(taskVo, nowTime);
        }
        return Boolean.TRUE;
    }


    /**
     * 记录日志
     */
    private void saveInStanceLog(IsfdDatasourceDataIntegrationTaskVo taskVo) {
        IsfdDatasourceTaskInstanceLog instanceLog = new IsfdDatasourceTaskInstanceLog();
        instanceLog.setTaskId(taskVo.getId());
        //具体字段的赋值还需要seatunnel工具集成的情况作为入参
        isfdDatasourceTaskInstanceLogMapper.insert(instanceLog);
    }

    /**
     * 执行seatunnel工具，开始集成数据
     */
    public void useSeaTunnelTool(IsfdDatasourceDataIntegrationTaskVo taskVo, Date nowTime) {
        // 从taskVo获取调度任务vo的id
        Long taskSchedulingConfigId = taskVo.getDatasourceTaskSchedulingConfigVo().getId();
        // todo 执行seatunnel工具，开始集成数据

/**
 // todo 这段代码要放到 查询到 seatunnel 任务状态之后记录
 Date triggerTime = nowTime;
 IsfdDatasourceTaskStatusEnum isfdDatasourceTaskStatusEnum = null;
 // 修改调度配置状态，具体成功与否根据调度情况
 if (IsfdDatasourceSyncTypeEnum.OFFLINE_SYNC.getValue().equals(taskVo.getSyncType())) {
 LambdaQueryWrapper<IsfdDatasourceTaskSchedulingConfig> lqw = new LambdaQueryWrapper<>();
 lqw.eq(IsfdDatasourceTaskSchedulingConfig::getId, taskSchedulingConfigId);
 IsfdDatasourceTaskSchedulingConfig taskSchedulingConfig = new IsfdDatasourceTaskSchedulingConfig();
 taskSchedulingConfig.setStatus(IsfdDatasourceTaskSchedulingConfigStatusEnum.NOT_RUNNING.getValue());
 // 设置下一次触发时间
 triggerTime = this.findTriggerTime(nowTime, taskSchedulingConfigId);
 taskSchedulingConfig.setTriggerTime(triggerTime);
 baseMapper.update(taskSchedulingConfig, lqw);
 isfdDatasourceTaskStatusEnum = IsfdDatasourceTaskStatusEnum.SCHEDULE_SUCCESS;
 } else {
 isfdDatasourceTaskStatusEnum = IsfdDatasourceTaskStatusEnum.SYNC_SUCCESS;
 }
 // 修改任务状态，具体成功与否根据调度情况
 iIsfdDatasourceDataIntegrationTaskService.updateStatusAndTriggerTime(isfdDatasourceTaskStatusEnum.getValue(), taskVo.getId(), triggerTime);
 */
        // 记录日志
        saveInStanceLog(taskVo);
    }

    private LambdaQueryWrapper<IsfdDatasourceTaskSchedulingConfig> buildQueryWrapper(IsfdDatasourceTaskSchedulingConfigBo bo) {
        Map<String, Object> params = bo.getParams();
        LambdaQueryWrapper<IsfdDatasourceTaskSchedulingConfig> lqw = Wrappers.lambdaQuery();
        lqw.eq(bo.getTaskId() != null, IsfdDatasourceTaskSchedulingConfig::getTaskId, bo.getTaskId());
        lqw.eq(bo.getSchedulingType() != null, IsfdDatasourceTaskSchedulingConfig::getSchedulingType, bo.getSchedulingType());
        lqw.eq(bo.getSchedulingCycle() != null, IsfdDatasourceTaskSchedulingConfig::getSchedulingCycle, bo.getSchedulingCycle());
        lqw.eq(StringUtils.isNotBlank(bo.getCycleUnit()), IsfdDatasourceTaskSchedulingConfig::getCycleUnit, bo.getCycleUnit());
        lqw.eq(bo.getStartTime() != null, IsfdDatasourceTaskSchedulingConfig::getStartTime, bo.getStartTime());
        lqw.eq(bo.getEndTime() != null, IsfdDatasourceTaskSchedulingConfig::getEndTime, bo.getEndTime());
        lqw.eq(bo.getEffectiveStartTime() != null, IsfdDatasourceTaskSchedulingConfig::getEffectiveStartTime, bo.getEffectiveStartTime());
        lqw.eq(bo.getEffectiveEndTime() != null, IsfdDatasourceTaskSchedulingConfig::getEffectiveEndTime, bo.getEffectiveEndTime());
        lqw.eq(bo.getTriggerTime() != null, IsfdDatasourceTaskSchedulingConfig::getTriggerTime, bo.getTriggerTime());
        lqw.eq(bo.getStatus() != null, IsfdDatasourceTaskSchedulingConfig::getStatus, bo.getStatus());
        lqw.eq(bo.getIsDeleted() != null, IsfdDatasourceTaskSchedulingConfig::getIsDeleted, bo.getIsDeleted());
        return lqw;
    }

}
